package net.bwie.realtime.jtp.dwd.trade.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.dwd.trade.function.IsFanClubProcessFunction;
import net.bwie.realtime.jtp.dwd.trade.function.IsNewCommentProcessFunction;
import net.bwie.realtime.jtp.dwd.trade.function.IsFirstOrderProcessFunction;
import net.bwie.realtime.jtp.dwd.trade.function.IsNewFanProcessFunction;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class UserTagJob {
    public static void main(String[] args) throws Exception {

        //1-执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2-数据源
        DataStream<String> eventStream1 = KafkaUtil.consumerKafka(env, "dwd_live_event_log");
//        eventStream1.print("日志表");
        DataStream<String> orderStream1 = KafkaUtil.consumerKafka(env, "dwd_trade_order_detail");
//        orderStream1.print("订单表");
        DataStream<String> relationStream1 = KafkaUtil.consumerKafka(env, "dwd_user_fan_relation");
//        relationStream1.print("粉丝主播关系表");

        //3-数据转换
        DataStream<String> eventStream = processLog(eventStream1,orderStream1,relationStream1);
        eventStream.print();

        //5-触发执行
        env.execute("UserTagJob");
    }

    private static DataStream<String> processLog(DataStream<String> eventStream1,
                                                 DataStream<String> orderStream1,
                                                 DataStream<String> relationStream1) {

        //1-用户是否首次下单
        DataStream<String> firstOrderStream= processIsFirstOrder(orderStream1);
        KafkaUtil.producerKafka(firstOrderStream,"dwd_user_first_order_detail");
        //2-新增粉丝团
        DataStream<String> newFansTeamStream= processIsNewFansTeam(relationStream1);
        KafkaUtil.producerKafka(newFansTeamStream,"dwd_user_new_fans_team_detail");
        //3-是否是新粉丝
        DataStream<String> newFansStream= processIsNewFans(relationStream1);
        KafkaUtil.producerKafka(newFansStream,"dwd_user_new_fans_detail");
        //4-是否新增评论
        DataStream<String> newCommentStream= processIsNewComment(eventStream1);
        KafkaUtil.producerKafka(newCommentStream,"dwd_user_new_comment_detail");
        return newCommentStream;
    }

    /**
     * 首次用户评论
     * @param stream
     * @return
     */
    private static DataStream<String> processIsNewComment(DataStream<String> stream) {
        KeyedStream<String, String> IsNewComment = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                /*
                {
                  "leave_time": "2025-08-26 10:24:25",
                  "live_room_id": 1001,
                  "anchor_id": 888,
                  "session_id": "ses_491484",
                  "sku_id": "sku_1007",
                  "source_type": "橱窗",
                  "enter_time": "2025-08-26 10:23:44",
                  "duration_sec": 41,
                  "platform": "抖音",
                  "event_id": "evt_359911",
                  "event_type": "评论",
                  "user_id": "usr_73389",
                  "live_session_id": "live_20250826_888_001",
                  "sku_name": "索尼降噪耳机",
                  "category": "手机数码",
                  "region": "深圳"
                }
                 */
                JSONObject object = JSON.parseObject(s);
                String user_id = object.getString("user_id");
                String anchor_id = object.getString("anchor_id");
                String live_session_id = object.getString("live_session_id");
                return user_id+"_"+anchor_id+"_"+live_session_id;

            }
        });

        //b-状态编程，对is_new_comment进行校验修复
        DataStream<String> isNewCommentStream = IsNewComment.process(new IsNewCommentProcessFunction());

        //c-返回数据流
        return isNewCommentStream;
    }

    /**
     * 用户是否是新增粉丝
     * @param stream
     * @return
     */
    private static DataStream<String> processIsNewFans(DataStream<String> stream) {
        KeyedStream<String, String> isNew = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                /*
                {
                    "is_not_fan": 0,
                    "anchor_id": 888,
                    "user_id": "usr_10375",
                    "is_new": 1,
                    "live_session_id": "live_20250826_888_001",
                    "change_time": "2025-08-26 16:17:24",
                    "relation_id": "rel_554820"
                }
                 */
                //todo is_new=1为新关注主笔，即为新粉丝，反之不是
                return JSON.parseObject(s).getString("is_new");
            }
        });
        //b-状态编程，对is_first_order进行校验修复
        DataStream<String> isNewFanStream = isNew.process(new IsNewFanProcessFunction());

        //c-返回数据流
        return isNewFanStream;
    }

    /**
     * 用户是否在粉丝团中
     * @param stream
     * @return
     */
    private static DataStream<String> processIsNewFansTeam(DataStream<String> stream) {
        KeyedStream<String, String> isNotFan = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                /*
                {
                  "is_not_fan": 0,
                  "anchor_id": 888,
                  "user_id": "usr_10465",
                  "is_new": 1,
                  "live_session_id": "live_20250826_888_001",
                  "change_time": "2025-08-26 11:16:25",
                  "relation_id": "rel_945567"
                }
                 */
                //todo is_not_fan=1为新加入主播粉丝团，反之不是
                return JSON.parseObject(s).getString("is_not_fan");
            }
        });
        //b-状态编程，对is_first_order进行校验修复
        DataStream<String> isNewFanTeamStream = isNotFan.process(new IsFanClubProcessFunction());

        //c-返回数据流
        return isNewFanTeamStream;
    }

    /**
     * 判断用户是否首次下单
     * @param stream
     * @return
     */
    private static DataStream<String> processIsFirstOrder(DataStream<String> stream) {
        KeyedStream<String, String> isFirstOrder = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                /*
                {
                  "live_room_id": 1001,
                  "anchor_id": 888,
                  "sku_id": "sku_1001",
                  "order_time": "2025-08-26 10:24:33",
                  "event_id": "evt_526319",
                  "user_id": "usr_77044",
                  "order_cnt": 2,
                  "is_first_order": 1,
                  "order_amount": 11139.68,
                  "live_session_id": "live_20250826_888_001",
                  "sku_name": "iPhone 16 Pro Max",
                  "category": "手机数码",
                  "order_id": "ord_348559",
                  "status": "finished"
                }
                 */
                //todo 如果is_first_order=1 则表示该用户是首次下单，反之不是
                return JSON.parseObject(s).getString("is_first_order");
            }
        });
        //b-状态编程，对is_first_order进行校验修复
        DataStream<String> isFirstOrderStream = isFirstOrder.process(new IsFirstOrderProcessFunction());

        //c-返回数据流
        return isFirstOrderStream;
    }
}
